package cn.ant.Client;

import cn.ant.utils.ChannelUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.MessageProperties;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.time.LocalTime;
import java.util.Map;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;


/**
 * 生产者（生产任务）
 *
 * @author Anhui AntLaddie <a href="https://juejin.cn/user/4092805620185316">(掘金蚂蚁小哥)</a>
 * @version 1.0
 **/
public class Producer {

    // 通过日志管理器获取Logger对象
    static Logger logger = LogManager.getLogger(Producer.class);
    // 单个发布确认
    public static final String SINGLE_RELEASE_CONFIRMATION = "singleReleaseConfirmation";
    // 批量确认
    public static final String BATCH_CONFIRMATION = "batchConfirmation";
    // 异步发布确认
    public static final String ASYNC_RELEASE_CONFIRMATION = "asyncReleaseConfirmation";

    /***
     * 调用 单个发布确认，批量发布确认，异步发布确认
     */
    public static void main(String[] args) throws IOException, InterruptedException {

        // 单个发布确认调用
        // singleReleaseConfirmation();
        // 批量发布确认
        // batchConfirmation();
        // 异步发布确认
        asyncReleaseConfirmation();
    }


    /***
     * 单个发布确认
     */
    public static void singleReleaseConfirmation() throws IOException, InterruptedException {
        // 通过工具类获取一个信道
        Channel channel = ChannelUtil.getChannel();
        // 声明一个队列
        channel.queueDeclare(SINGLE_RELEASE_CONFIRMATION, true, false, false, null);
        // 开启发布确认功能！！！
        channel.confirmSelect();
        // 记录开始时间
        long start = System.currentTimeMillis();

        // 发送消息
        for (int i = 1; i <= 100; i++) {
            byte[] msg = ("这是一个编号为：" + i + " 的待处理的消息").getBytes(StandardCharsets.UTF_8);
            // 依次发送每一条消息到消息队列中
            channel.basicPublish("", SINGLE_RELEASE_CONFIRMATION, MessageProperties.PERSISTENT_BASIC, msg);
            // 验证是否发送成功（阻塞等待队列返回的确认消息）
            // 发送三秒后没得到回复将断定未发送过去
            // boolean waitTime = channel.waitForConfirms(3000);
            // 将一直等待刚才发送的消息是否被RabbitMQ确认
            boolean waitSuccess = channel.waitForConfirms();
            if (waitSuccess) {
                // 如果选择的是info日志级别，所花的时间会高一点，因为日志级别为info，打印耗费一些时间
                logger.debug("[单个发布确认]发送成功了，已发送到RabbitMQ队列中，发送信息为：{}", i);
            } else {
                logger.debug("[单个发布确认]发送失败，未收到RabbitMQ确认信息，发送信息为：{}", i);
            }
        }
        long end = System.currentTimeMillis();
        logger.info("生产者消息发送完成【单个发布确认，用时：{}】", end - start);
    }


    /***
     * 批量发布确认
     */
    public static void batchConfirmation() throws IOException, InterruptedException {
        // 通过工具类获取一个信道
        Channel channel = ChannelUtil.getChannel();
        // 声明一个队列
        channel.queueDeclare(BATCH_CONFIRMATION, true, false, false, null);
        // 开启发布确认功能！！！
        channel.confirmSelect();
        // 记录开始时间
        long start = System.currentTimeMillis();

        // 定义每次批量处理多少消息进行确认
        int batchNumber = 10;
        // 发送的消息
        for (int i = 1; i <= 100; i++) {
            byte[] msg = ("这是一个编号为：" + i + " 的待处理的消息").getBytes(StandardCharsets.UTF_8);
            // 依次发送每一条消息到消息队列中
            channel.basicPublish("", BATCH_CONFIRMATION, MessageProperties.PERSISTENT_BASIC, msg);
            // 验证是否发送成功，对10求余，代表每发送10条消息会批量看看是否成功（等待确认）
            if (i % batchNumber == 0) {
                boolean waitSuccess = channel.waitForConfirms();
                // 计算，到时候日志需要打印
                String str = (i - (batchNumber - 1)) + " ~ " + i;
                // 判断是否批量确认成功
                if (waitSuccess) {
                    // 如果选择的是info日志级别，所花的时间会高一点，因为日志级别为info，打印耗费一些时间
                    logger.info("[批量发布确认]发送成功了，已发送到RabbitMQ队列中，发送信息范围：{}", str);
                } else {
                    logger.info("[批量发布确认]发送失败，未收到RabbitMQ确认信息，发送信息范围：{}", str);
                }
            }
        }
        long end = System.currentTimeMillis();
        logger.info("生产者消息发送完成【批量发布确认，用时：{}】", end - start);
    }

    /***
     * 异步发布确认
     */
    public static void asyncReleaseConfirmation() throws IOException {
        // 通过工具类获取一个信道
        Channel channel = ChannelUtil.getChannel();
        // 声明一个队列
        channel.queueDeclare(ASYNC_RELEASE_CONFIRMATION, true, false, false, null);
        // 开启发布确认功能！！！
        channel.confirmSelect();

        // ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ Start
        // 线程安全有序的一个哈希表Map，适用于高并发的情况
        // 1.轻松的将序号与消息进行关联 2.轻松批量删除条目 只要给到序列号 3.支持并发访问
        ConcurrentSkipListMap<Object, Object> outstandingConfirms = new ConcurrentSkipListMap<>();

        // 消息确认成功，回调函数
        // 回调成功后把集合中的确认数据删除，最终没删除的就代表是失败的
        ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
            // 是否为批量确认，当数据回调快的话是很有可能走批量确认的，具体看队列返回的回调情况
            if (multiple) {
                // 获取集合开头到deliveryTag数的数据记录，并组成新的集合，true代表包含当前这个数，false不包含
                ConcurrentNavigableMap<Object, Object> navigableMap =
                        outstandingConfirms.headMap(deliveryTag, true);
                navigableMap.clear();
            } else {
                // 若不批量则一个一个删除集合中的数据
                outstandingConfirms.remove(deliveryTag);
            }
            logger.info("回调成功的数据：{}   是否批量确认：{}", deliveryTag, multiple);
        };

        // 消息确认失败，回调函数
        ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
            logger.info("回调失败的数据：{}", deliveryTag);
        };

        // 添加监听器，监听返回(监听器一定要在发送消息之前就创建和监听)
        // 参数1：回调成功方法 参数2：回调失败方法
        channel.addConfirmListener(ackCallback, nackCallback);
        //~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ End

        // 记录开始时间
        long start = System.currentTimeMillis();
        // 发送的消息
        for (int i = 1; i <= 100; i++) {
            byte[] msg = ("这是一个编号为：" + i + " 的待处理的消息").getBytes(StandardCharsets.UTF_8);
            // 把要发送的数据记录到集合中
            outstandingConfirms.put(channel.getNextPublishSeqNo(), msg);
            // 依次发送每一条消息
            channel.basicPublish("", ASYNC_RELEASE_CONFIRMATION,
                    MessageProperties.PERSISTENT_TEXT_PLAIN, msg);
        }
        long end = System.currentTimeMillis();
        logger.info("生产者消息发送完成【异步发布确认，用时：{}】", end - start);

        // 到最后就可以处理发送确认失败的数据了
        for (Map.Entry<Object, Object> outstandingConfirm : outstandingConfirms.entrySet()) {
            // 在这里可以重新修改投递啥的
            logger.debug("处理失败的消息：{}", outstandingConfirm);
        }
    }
}